package beautiful.butterfly.drds.data_exchange.taskgroup.runner;


import beautiful.butterfly.drds.data_exchange.data_exchange.exchanger.Consume;
import beautiful.butterfly.drds.data_exchange.plugin.AbstractTaskPlugin;
import beautiful.butterfly.drds.data_exchange.read_and_write_data.AbstractWriter;
import beautiful.butterfly.drds.data_exchange.report.Messages;
import beautiful.butterfly.drds.data_exchange.statistics.Phase;
import beautiful.butterfly.drds.data_exchange.statistics.RunningInfo;
import org.apache.commons.lang3.Validate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


public class WriteTask extends Task implements Runnable
{

    private static final Logger log = LoggerFactory
            .getLogger(WriteTask.class);

    private Consume consume;

    public WriteTask(AbstractTaskPlugin abstractTaskPlugin)
    {
        super(abstractTaskPlugin);
    }

    public void setConsume(Consume consume)
    {
        this.consume = consume;
    }

    //@Override
    public void run()
    {
        Validate.isTrue(this.consume != null);

        AbstractWriter.AbstractTask abstractTask = (AbstractWriter.AbstractTask) this.getTaskPlugin();

        RunningInfo runningInfo = new RunningInfo(getTaskGroupId(), getTaskId(), Phase.wait_read_time);
        try
        {
            runningInfo.start();
            //
            log.debug("task write starts to do init ...");
            RunningInfo initRunningInfo = new RunningInfo(getTaskGroupId(), getTaskId(), Phase.write_task_init);
            initRunningInfo.start();
            abstractTask.init();
            initRunningInfo.end();

            log.debug("task write starts to do prepare ...");
            RunningInfo prepareRunningInfo = new RunningInfo(getTaskGroupId(), getTaskId(), Phase.write_task_prepare);
            prepareRunningInfo.start();
            abstractTask.prepare();
            prepareRunningInfo.end();
            log.debug("task write starts to write ...");

            RunningInfo dataRunningInfo = new RunningInfo(getTaskGroupId(), getTaskId(), Phase.write_task_start);
            dataRunningInfo.start();
            abstractTask.startWrite(consume);

            dataRunningInfo.addCount(Messages.getTotalReadRecords(super.getMessage()));
            dataRunningInfo.addSize(Messages.getTotalReadBytes(super.getMessage()));
            dataRunningInfo.end();

            log.debug("task write starts to do post ...");
            RunningInfo postRunningInfo = new RunningInfo(getTaskGroupId(), getTaskId(), Phase.write_task_post);
            postRunningInfo.start();
            abstractTask.post();
            postRunningInfo.end();

            super.markSuccess();
        } catch (Throwable e)
        {
            log.error("AbstractWriter Runner Received Exceptions:", e);
            super.markFail(e);
        } finally
        {
            log.debug("task write starts to do destroy ...");
            RunningInfo destroyRunningInfo = new RunningInfo(getTaskGroupId(), getTaskId(), Phase.write_task_destroy);
            destroyRunningInfo.start();
            super.destroy();
            destroyRunningInfo.end();
            runningInfo.end(super.getMessage().getLong(Messages.wait_reader_time));
        }
    }

    public boolean supportFailOver()
    {
        AbstractWriter.AbstractTask abstractTask = (AbstractWriter.AbstractTask) this.getTaskPlugin();
        return abstractTask.supportFailOver();
    }

    public void shutdown()
    {
        consume.shutdown();
    }
}
